# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor
from jms_dm_door_pick.dm.dm_thd_door_pick_up_summary_dt import jms_dm__dm_thd_door_pick_up_summary_dt

label = f"dm_thd_door_pick_up_summary_dt_{int(time())}"
timeout = timedelta(minutes=30).seconds
doris_jms_dm__dm_thd_door_pick_up_summary_dt = RichSqlSensor(
    task_id='doris_jms_dm__dm_thd_door_pick_up_summary_dt',
    pool='broker_load_pool',
    email=['suning@jtexpress.com','yl_bigdata@yl-scm.com'],
    conn_id='doris',
    pre_sql=f"""
    TRUNCATE TABLE jms_dm.dm_thd_door_pick_up_summary_dt PARTITION (
        p{{{{ execution_date | cst_ds_nodash }}}}
    );
    LOAD LABEL jms_dm.{label} (
        DATA INFILE(
            "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_thd_door_pick_up_summary_dt/dt={{{{ execution_date | cst_ds }}}}/*"
        )
        INTO TABLE dm_thd_door_pick_up_summary_dt
        FORMAT AS 'PARQUET'(       
             serach_type
            ,site_code
            ,site_name
            ,agent_code
            ,agent_name
            ,virt_code
            ,virt_name
            ,order_num
            ,recept_num
            ,pick_num
            ,cancel_num
            ,overtime_pick_num
            ,timely_pick_num
            ,break_num
            ,courier_change_num
            ,change_other_num
            ,complaint_num
            ,complaint_again_num
            ,send_back_num
            ,should_collect_num
            ,timely_collect_num
            ,sign_num
            ,pricing_num
            ,timely_pricing_num
            ,hj_fine_amt
            ,qx_fine_amt
            ,wdts_fine_amt
            ,lsbjs_fine_amt
            ,fine_amt
            ,timely_pick_rate
            ,cancel_rate
            ,break_rate
            ,courier_change_rate
            ,complaint_rate
            ,phone_conn_rate
            ,pun_phone_conn_rate
            ,timely_collect_rate
            ,sign_rate
            ,timely_pricing_rate
            ,wd_complaint_rate
            ,hj_score
            ,ls_score
            ,qx_score
            ,wdts_score
            ,ts_score
            ,total_score
            ,month_hj_fine_amt
            ,month_qx_fine_amt
            ,month_wdts_fine_amt
            ,month_lsbjs_fine_amt
            ,month_fine_amt
            ,month_timely_pricing_rate
            ,month_timely_collect_rate
            ,month_wd_complaint_rate
            ,month_cancel_rate
            ,month_complaint_num
            ,month_hj_score
            ,month_ls_score
            ,month_wdts_score
            ,month_qx_score
            ,month_ts_score
            ,month_total_score
            ,month_total_not_reach_cnt
            ,bdb_network_cnt
            ,month_bdb_network_cnt
        )
        columns from path as (dt)
        set(
              serach_type                    = serach_type
             ,site_code                      = site_code
             ,site_name                      = site_name
             ,agent_code                     = agent_code
             ,agent_name                     = agent_name
             ,virt_code                      = virt_code
             ,virt_name                      = virt_name
             ,order_num                      = order_num
             ,recept_num                     = recept_num
             ,pick_num                       = pick_num
             ,cancel_num                     = cancel_num
             ,overtime_pick_num              = overtime_pick_num
             ,timely_pick_num                = timely_pick_num
             ,break_num                      = break_num
             ,courier_change_num             = courier_change_num
             ,change_other_num               = change_other_num
             ,complaint_num                  = complaint_num
             ,complaint_again_num            = complaint_again_num
             ,send_back_num                  = send_back_num
             ,should_collect_num             = should_collect_num
             ,timely_collect_num             = timely_collect_num
             ,sign_num                       = sign_num
             ,pricing_num                    = pricing_num
             ,timely_pricing_num             = timely_pricing_num
             ,hj_fine_amt                    = hj_fine_amt
             ,qx_fine_amt                    = qx_fine_amt
             ,wdts_fine_amt                  = wdts_fine_amt
             ,lsbjs_fine_amt                 = lsbjs_fine_amt
             ,fine_amt                       = fine_amt
             ,timely_pick_rate               = timely_pick_rate
             ,cancel_rate                    = cancel_rate
             ,break_rate                     = break_rate
             ,courier_change_rate            = courier_change_rate
             ,complaint_rate                 = complaint_rate
             ,phone_conn_rate                = phone_conn_rate
             ,pun_phone_conn_rate            = pun_phone_conn_rate
             ,timely_collect_rate            = timely_collect_rate
             ,sign_rate                      = sign_rate
             ,timely_pricing_rate            = timely_pricing_rate
             ,wd_complaint_rate              = wd_complaint_rate
             ,hj_score                       = hj_score
             ,ls_score                       = ls_score
             ,qx_score                       = qx_score
             ,wdts_score                     = wdts_score
             ,ts_score                       = ts_score
             ,total_score                    = total_score
             ,month_hj_fine_amt              = month_hj_fine_amt
             ,month_qx_fine_amt              = month_qx_fine_amt
             ,month_wdts_fine_amt            = month_wdts_fine_amt
             ,month_lsbjs_fine_amt           = month_lsbjs_fine_amt
             ,month_fine_amt                 = month_fine_amt
             ,month_timely_pricing_rate      = month_timely_pricing_rate
             ,month_timely_collect_rate      = month_timely_collect_rate
             ,month_wd_complaint_rate        = month_wd_complaint_rate
             ,month_cancel_rate              = month_cancel_rate
             ,month_complaint_num            = month_complaint_num
             ,month_hj_score                 = month_hj_score
             ,month_ls_score                 = month_ls_score
             ,month_wdts_score               = month_wdts_score
             ,month_qx_score                 = month_qx_score
             ,month_ts_score                 = month_ts_score
             ,month_total_score              = month_total_score
             ,month_total_not_reach_cnt      = month_total_not_reach_cnt
             ,bdb_network_cnt                = bdb_network_cnt
             ,month_bdb_network_cnt          = month_bdb_network_cnt
             ,dt                             = dt
        )
    )
    WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dm WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dm WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_jms_dm__dm_thd_door_pick_up_summary_dt << jms_dm__dm_thd_door_pick_up_summary_dt
